package net.cassandraview.db;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

/**
 * Pool of connections to cassandra instance. For now - just a simple implementation with count of initialized
 * by this pool connections.
 * 
 * Yes, i know about existing pool libraries, but don't want to have unneeded dependencies ;)
 * 
 * User: nagakhl
 * Date: 21.01.12
 * Time: 21:38
 */
public class ConnectionPool <T> {
    
    private Logger log = Logger.getLogger(getClass().getName());
    
    private int initial, max, count = 0;
    
    private Lock lock;

    private Condition notEmpty;

    private List<T> availableConnections;
    
    private ConnectionProducer<? extends T> producer;
    
    ConnectionPool(int initial, int max, ConnectionProducer<? extends T> producer) {
        this.initial = initial;
        this.max = max;
        lock = new ReentrantLock();
        notEmpty = lock.newCondition();
        availableConnections = new ArrayList<T>(max);
        while (count < this.initial) {
            availableConnections.add(producer.initalizeConnection());
            count++;
        }
    }
    
    public T getConnection() {
        lock.lock();
        try {
            if (availableConnections.isEmpty()) {
                if (count < max) {
                    count++;
                    return producer.initalizeConnection();
                } else {
                    while (availableConnections.isEmpty())
                        notEmpty.await();
                    return availableConnections.remove(availableConnections.size() - 1);
                }
            } else {
                return availableConnections.remove(availableConnections.size() - 1);
            }
        } catch (InterruptedException e) {
            log.severe("Interrupted while waiting for connections to release.");
        } finally {
            lock.unlock();
        }
        return null;
    }

    public void returnConnection(T connection) {
        lock.lock();
        try {
            availableConnections.add(connection);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }
    
}
